package org.example.rocketmq.msg_type.batch;

import org.apache.commons.lang3.ThreadUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.time.Duration;
import java.time.LocalTime;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class ProducerBatch {

    public static void main(String[] args) throws Exception {
        producer();
    }


    /**
     * 批量消息是指将多条消息合并成一个批量消息，一次发送出去，这样的好处是可以减少网络IO，提升吞吐量
     * 批量消息的使用限制：
     * 消息大小不能超过4M, 虽然源码注释不能超1M, 但是实际使用不超过4M即可。平衡整体的性能，建议保持1M左右；
     * 相同的 topic
     * 相同的 waitStoreMsgOK
     * 不能是延迟消息/事务消息等
     */
    public static void producer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("producer_batch_group_01");
        producer.setNamesrvAddr("rocketmq.jun.cn:9876");
        producer.start();

        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 100000; i++) {
            Message message = new Message("TopicTest", "TagA", ("Hello RocketMQ Batch" + i + "_step").getBytes(RemotingHelper.DEFAULT_CHARSET));
            messages.add(message);
        }

        { // 分批发送
            ListSplitter listSplitter = new ListSplitter(messages);
            int i = 0;
            while (listSplitter.hasNext()) {
                SendResult result = producer.send(listSplitter.next());
                System.out.printf(LocalTime.now() + "消息发送成功 " + (i++) + " %s%n", result);
            }
        }

        producer.shutdown();
    }

}


class ListSplitter implements Iterator<List<Message>> {
    private static final int SIZE_LIMIT = 10 * 1000; // 单个 messages 最大字节数   1M
    private final List<Message> messages; // 每次处理的集合数
    private int currentIndex; // 当前的索引

    ListSplitter(List<Message> messages) {
        this.messages = messages;
    }

    @Override
    public boolean hasNext() {
        return currentIndex < messages.size();
    }

    @Override
    public List<Message> next() {
        int nextIndex = currentIndex;
        int totalSize = 0;

        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);

            int messageSize = message.getBody().length + message.getTopic().length();
            {
                Map<String, String> properties = message.getProperties();
                Iterator<Map.Entry<String, String>> iterator = properties.entrySet().iterator();
                while (iterator.hasNext()) {
                    Map.Entry<String, String> next = iterator.next();
                    messageSize += next.getKey().length() + next.getValue().length();
                }
                messageSize = messageSize + 20;
            }

            if (messageSize > SIZE_LIMIT) {
                // 第一次发送就超出了限制，直接跳过这条消息继续扫描
                if (nextIndex - currentIndex == 0) {
                    nextIndex++;
                }
                break;
            }

            // 如果当前发送列表已经超出限制，暂停添加消息
            if (messageSize + totalSize > SIZE_LIMIT) {
                break;
            } else {
                totalSize += messageSize;
            }
        }

        List<Message> messageList = this.messages.subList(currentIndex, nextIndex);
        currentIndex = nextIndex;

        return messageList;
    }
}
